package com.jy.datapipeline.export.splitkey.split.impl;

import com.jy.datapipeline.common.util.RangeSplitUtil;
import com.jy.datapipeline.export.common.constants.SplitKeyExecuteConstant;
import com.jy.datapipeline.export.splitkey.split.Split;
import com.jy.datapipeline.export.splitkey.handler.SplitKeyTaskHandler;
import com.jy.datapipeline.export.splitkey.utils.AssertUtil;
import org.springframework.stereotype.Component;

import java.math.BigInteger;
import java.util.Map;

/**
 * @类名 RoughSplit
 * @描述 TODO
 * @作者 白条君
 * @创建日期 2023/8/25 14:36
 * @版本 1.0
 */
@Component
public class RoughSplit implements Split {

    @Override
    public void execute(SplitKeyTaskHandler splitKeyTaskHandler) throws Exception {
        Map<String, Object> res = splitKeyTaskHandler.initSplitKeyTask();
        int pieces = (int) res.get(SplitKeyExecuteConstant.PIECES);
        int MinSplitPKType = (int) res.get(SplitKeyExecuteConstant.SPLIT_KEY_TYPE);
        Object MaxSplitPK = res.get(SplitKeyExecuteConstant.MAX_SPLIT_KEY);
        Object MinSplitPK = res.get(SplitKeyExecuteConstant.MIN_SPLIT_KEY);

        if (AssertUtil.isNumberType(MinSplitPKType)) {
            if (String.valueOf(MaxSplitPK).equals(String.valueOf(MinSplitPK))) {
                MaxSplitPK = MaxSplitPK + "1";
            }

            BigInteger maxSplitPk = new BigInteger(String.valueOf(MaxSplitPK));
            BigInteger minSplitPk = new BigInteger(String.valueOf(MinSplitPK));

            BigInteger[] ResultSplitPk = RangeSplitUtil.doBigIntegerSplit(minSplitPk, maxSplitPk, pieces);
            if (ResultSplitPk.length > 0) {
                for (int i = 0; i < ResultSplitPk.length - 1; i++) {
                    BigInteger startValue = ResultSplitPk[i];
                    BigInteger endValue = ResultSplitPk[i + 1];
                    String datafileSuffix = i < 10 ? "000" + (i + 1) : "00" + (i + 1);

                    boolean isFinish = i == ResultSplitPk.length - 2;

                    splitKeyTaskHandler.executeSplitKey(startValue, endValue, datafileSuffix, isFinish);
                }
            }
        }

        if (AssertUtil.isNotNumberType(MinSplitPKType)) {
            String[] ResultSplitPk = RangeSplitUtil.doAsciiStringSplit(String.valueOf(MinSplitPK), String.valueOf(MaxSplitPK), pieces);
            if (ResultSplitPk.length > 0) {
                for (int i = 0; i < ResultSplitPk.length - 1; i++) {
                    String startValue = ResultSplitPk[i];
                    String endValue = ResultSplitPk[i + 1];
                    if (startValue.equals(endValue)) {
                        endValue = endValue + "1";
                    }
                    boolean isFinish = i == ResultSplitPk.length - 2;
                    String datafileSuffix = i < 10 ? "000" + (i + 1) : "00" + (i + 1);
                    splitKeyTaskHandler.executeSplitKey("'" + startValue + "'",
                            "'" + endValue + "'", datafileSuffix, isFinish);
                }
            }
        }
    }
}
